package com.auto.base.util;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.auto.base.model.MqItem;
import com.auto.base.params.BaseAutoConfig;
import io.qameta.allure.Step;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.util.Random;

/**
 * @author Chance.W
 * @Description mq连接
 * @date 2021/4/16-22:53
 */
public class MqTemplate {
    private final Logger log = LoggerFactory.getLogger(this.getClass());
    DefaultMQProducer producer = new DefaultMQProducer("auto-test-roducer");
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("auto-test-consumer");

    public MqTemplate(BaseAutoConfig autoConfig) {
        producer.setNamesrvAddr(autoConfig.getMqNamesrvAddr());
        consumer.setNamesrvAddr(autoConfig.getMqNamesrvAddr());
    }

    @Step("发送MQ消息")
    public SendResult sendMsg(MqItem mqItem) throws MQClientException, ParseException, RemotingException, InterruptedException, MQBrokerException {
        producer.start();
        log.info("send message is {}", mqItem);
        Message msg = new Message(mqItem.getTopic(), mqItem.getTags(), mqItem.getMessage().getBytes(StandardCharsets.UTF_8));
        msg.setKeys(SnowflakeIdWorker.generateId().toString());
        SendResult sendResult = producer.send(msg);
        log.info("end send message {}", sendResult);
        return sendResult;
    }

    @Step("消费MQ消息")
    public void consumerMsg(MqItem mqItem) throws MQClientException {
        consumer.subscribe(mqItem.getTopic(), "*");
        consumer.setInstanceName("auto-test");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            log.info("{} Receive New Messages: {}", Thread.currentThread().getName(), msgs);
            for (MessageExt messageExt : msgs) {
                String messageBody = new String(messageExt.getBody());
                log.info("消费响应：msgId : {},  msgBody : {}", messageExt.getMsgId(), messageBody);//输出消息内容
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }
}
